<?php


class QueueTest {

    private static $hook_list = [];//队列任务
    private static $hooked = false;
    /**
     * 启动入口
     */
    function run() {
        $redis=new \Redis();
        $redis->connect('127.0.0.1', 6379);
        for ($i=0;$i<=10;$i++) $redis->lpush("list",$i);
        self::hook([$this, 'queue'], ['处理了']);
        echo '响应成功';
    }

    /**
     * 消耗队列
     * @param $a
     */
    function queue($a){
        $redis=new \Redis();
        $redis->connect('127.0.0.1', 6379);
        $i=0;
        $co=$redis->lRange ('list',0,-1);
        while ($i<count ($co)){
            $rdata=$redis->brPop("list",5);
            if($rdata){
                sleep(5);
                file_put_contents (__DIR__.'/jyy.log',$a.$rdata[1]."\n\r",FILE_APPEND);
            }
            $i++;
            if($i==$co) break;
        }
    }

    /**
     * hook函数fastcgi_finish_request执行
     * @param callback $callback
     * @param array $params
     */
    public static function hook($callback, $params) {
        self::$hook_list[] = ['callback' => $callback, 'params' => $params];
        if(self::$hooked == false) {
            self::$hooked = true;
            register_shutdown_function([__CLASS__, 'sync']);//程序退出之前会触发这个函数注册的函数
        }
    }

    /**
     * 由系统调用
     * @return void
     */
    public static function sync() {
        fastcgi_finish_request();//运行在FastCGI模式，是fpm提供的函数,windows是没有的,运行到这步，响应数据、关闭连接，php看似退出进程，实际不会结束后续代码的运行。
        set_time_limit(0);  //避免超时报错
        ini_set('memory_limit', '-1');  //避免内存不足
        if(empty(self::$hook_list)) exit();
        foreach(self::$hook_list as $hook) {
            $callback = $hook['callback'];
            $params = $hook['params'];
            call_user_func_array($callback, $params);
        }
    }
}

$run = new QueueTest();
$run->run();